package com.funny.rabbitmq.confirm;

import com.funny.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 开启confirm模式的方法：channel.confirmSelect();
 * 客户端实现confirm的方法有三种：
 * 普通模式：每发送一条消息，就会等待消费者的回应，极大的浪费性能。（不推荐）
 * 批量模式：批量发送消息，但是如果遇到发送的消息中有一条失败，那么这一批的所有消息都要重新发送，导致消息的大量重复发送。（不推荐）
 * 异步模式：异步的发送确认机制，不会阻塞当前队列的运行，适合实际开发使用。（推荐）
 */
@Slf4j
public class Producer {

    private final static String QUEUE_NAME = "hello";
    // 处理未确认的消息

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 开启发布确认
        channel.confirmSelect();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        long start = System.currentTimeMillis();


        /**
         * 线程安全有序的一个哈希表，
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        channel.addConfirmListener((deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            } else {
                outstandingConfirms.remove(deliveryTag);
            }
            log.info("确认的消息 deliveryTag={}, multiple={}", deliveryTag, multiple);
        }, ((deliveryTag, multiple) -> {
            log.info("未确认的消息 deliveryTag={}, multiple={}", deliveryTag, multiple);
        }));

        for (int i = 0; i < 1000; i++) {
            String msg = "hello world_" + i;
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            outstandingConfirms.put(channel.getNextPublishSeqNo(), msg);
            System.out.println(channel.getNextPublishSeqNo());

//            boolean flag = channel.waitForConfirms();
//            if (flag)
                System.out.println("消息发送成功");
        }
        long end = System.currentTimeMillis();
        System.out.println(end -start);
        // confirm:
        // 数据量      1000    100000
        // 不开启      37      3297
        //  开启      569
    }
}
